package com.ferry.lock.engine.zookeeper.menagerie;

import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by daiyong
 * 默认session管理器
 */
public class DefaultZKSessionManager implements ZKSessionManager {

	private static final Logger logger = LoggerFactory.getLogger(DefaultZKSessionManager.class);

	private List<ConnectionListener> listeners = Collections.synchronizedList(new ArrayList<ConnectionListener>());

	private ZooKeeper zk;
	private final String connectionString;
	private final int timeout;
	private final ExecutorService executor;
	private volatile boolean shutdown;
	private ZKSessionPoller poller;
	private final int zkSessionPollInterval;

	public DefaultZKSessionManager(String connectionString, int timeout) {
		this(connectionString,timeout, Executors.newSingleThreadExecutor(),-1);
	}

	public DefaultZKSessionManager(String connectionString, int timeout, int zkSessionPollInterval){
		this(connectionString,timeout,Executors.newSingleThreadExecutor(),zkSessionPollInterval);
	}

	public DefaultZKSessionManager(String connectionString, int timeout,ExecutorService executor) {
		this(connectionString,timeout,executor,-1);
	}

	public DefaultZKSessionManager(String connectionString, int timeout, ExecutorService executor, int zkSessionPollInterval){
		this.connectionString = connectionString;
		this.timeout = timeout;
		this.executor = executor;
		this.zkSessionPollInterval = zkSessionPollInterval;
	}

	@Override
	public ZooKeeper getZooKeeper() {
		if (shutdown) {
			throw new IllegalStateException("连接已关闭");
		}

		if (zk == null || zk.getState() == ZooKeeper.States.CLOSED) {
			try {
				zk = new ZooKeeper(connectionString , timeout, new SessionWatcher(this));
			} catch (IOException e) {
				logger.error("创建zk异常", e);
				throw new RuntimeException(e);
			}

			if (zkSessionPollInterval > 0) {
				if (poller != null) {
					poller.stopPolling();
				}
				poller = new ZKSessionPoller(zk, zkSessionPollInterval, new SessionPollListener(zk, this));
				poller.startPolling();
			}
		} else {
			zk.sync("/", new AsyncCallback.VoidCallback() {
				@Override
				public void processResult(int rc, String path, Object ctx) {
					logger.info("链接zookeeper, rc:{}", rc);
				}
			}, this);
		}

		return zk;
	}

	@Override
	public void closeSession() {
		try {
			if (zk != null) {
				try {
					zk.close();
				} catch (InterruptedException e) {
					throw new RuntimeException(e);
				}
			}
		} finally {
			executor.shutdown();
			shutdown = true;
		}
	}

	@Override
	public void addConnectionListener(ConnectionListener listener) {
		listeners.add(listener);
	}

	@Override
	public void removeConnectionListener(ConnectionListener listener) {
		listeners.remove(listener);
	}

	private void notifyListeners(WatchedEvent event) {
		notifyState(event.getState());
	}

	private void notifyState(final Watcher.Event.KeeperState state) {
		//异步触发监听者
		executor.submit(new Runnable() {
			@Override
			public void run() {
				//超时
				if (state == Watcher.Event.KeeperState.Expired) {
					for(ConnectionListener listener : listeners){
						listener.expired();
					}
				} else if (state == Watcher.Event.KeeperState.SyncConnected) {
					for(ConnectionListener listener : listeners){
						listener.syncConnected();
					}
				} else if (state == Watcher.Event.KeeperState.Disconnected) {
					for(ConnectionListener listener : listeners){
						listener.disconnected();
					}
				}
			}
		});
	}

	private class SessionPollListener extends ConnectionListenerSkeleton {
		private final ZooKeeper zk;
		private final DefaultZKSessionManager sessionManager;

		public SessionPollListener(ZooKeeper zk, DefaultZKSessionManager sessionManager) {
			this.zk = zk;
			this.sessionManager = sessionManager;
		}

		@Override
		public void expired() {
			sessionManager.notifyState(Watcher.Event.KeeperState.Expired);
			try {
				zk.close();
			} catch (InterruptedException e) {
				logger.warn("An InterruptedException was detected while attempting to close a ZooKeeper instance; ignoring because we're shutting it down anyway");
			}
		}
	}

	private class SessionWatcher implements Watcher {

		private final DefaultZKSessionManager manager;

		public SessionWatcher(DefaultZKSessionManager manager) {
			this.manager = manager;
		}

		@Override
		public void process(WatchedEvent event) {
			manager.notifyListeners(event);
		}
	}
}
